-------------------------------------------------------------------------------------------------------------


# 导入必要模块
from langchain.document_loaders import DirectoryLoader, JSONLoader, MarkdownLoader
from langchain.text_splitter import CharacterTextSplitter

# 定义数据加载和预处理模块
class DataLoaderModule:
    def __init__(self, directory_path, chunk_size=500, chunk_overlap=50):
        """
        初始化数据加载模块，设置文件目录路径、分块大小和重叠。
        :param directory_path: 要加载的文件目录路径
        :param chunk_size: 文档分割的每个块大小
        :param chunk_overlap: 文档块之间的重叠大小
        """
        self.directory_path = directory_path
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.documents = []
        self.text_splitter = CharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)

    def load_text_documents(self):
        """加载文本文件"""
        loader = DirectoryLoader(path=self.directory_path, glob="**/*.txt", show_progress=True)
        raw_documents = loader.load()
        self.documents.extend(self.text_splitter.split_documents(raw_documents))
        print(f"加载的文本文件数：{len(raw_documents)}")

    def load_json_documents(self, json_path):
        """加载JSON文件"""
        loader = JSONLoader(file_path=json_path)
        json_documents = loader.load()
        split_docs = self.text_splitter.split_documents(json_documents)
        self.documents.extend(split_docs)
        print(f"加载的JSON文件条目数：{len(json_documents)}")

    def load_markdown_documents(self, markdown_path):
        """加载Markdown文件"""
        loader = MarkdownLoader(file_path=markdown_path)
        markdown_documents = loader.load()
        split_docs = self.text_splitter.split_documents(markdown_documents)
        self.documents.extend(split_docs)
        print(f"加载的Markdown文件数：{len(markdown_documents)}")

    def validate_and_clean_data(self):
        """数据验证和清洗：去除空内容，删除重复条目"""
        # 去除空内容
        self.documents = [doc for doc in self.documents if doc.page_content.strip()]
        
        # 去重处理
        unique_docs = {doc.page_content: doc for doc in self.documents}
        self.documents = list(unique_docs.values())
        
        print(f"清洗后剩余文档数：{len(self.documents)}")

    def load_all_documents(self, json_path, markdown_path):
        """加载所有文档：文本、JSON、Markdown"""
        self.load_text_documents()
        self.load_json_documents(json_path)
        self.load_markdown_documents(markdown_path)
        
        # 数据清洗
        self.validate_and_clean_data()

    def display_summary(self):
        """显示数据加载和处理的结果统计"""
        print(f"总文档数：{len(self.documents)}")
        for i, doc in enumerate(self.documents[:5]):  # 显示前5个文档内容概览
            print(f"\n文档{i + 1}内容预览:\n{doc.page_content[:200]}...")  # 只显示前200字符


# 模块运行测试
if __name__ == "__main__":
    # 初始化DataLoaderModule，指定文件路径
    data_loader = DataLoaderModule(directory_path="./data")

    # 加载所有文件，并执行数据清洗
    data_loader.load_all_documents(json_path="./data/sample.json", markdown_path="./data/sample.md")
    
    # 输出加载与处理结果的概览
    data_loader.display_summary()


-------------------------------------------------------------------------------------------------------------


# 导入必要模块
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import FAISS
import os

# 嵌入生成与存储模块
class EmbeddingStorageModule:
    def __init__(self, data_loader, faiss_index_path="faiss_index"):
        """
        初始化嵌入生成与存储模块，设置数据加载模块和FAISS索引存储路径。
        :param data_loader: 已加载和清洗的DataLoaderModule对象
        :param faiss_index_path: FAISS向量存储的路径
        """
        self.data_loader = data_loader  # 加载后的数据
        self.embeddings = OpenAIEmbeddings()  # 嵌入模型
        self.faiss_index_path = faiss_index_path  # FAISS向量存储路径
        self.vector_store = None  # 初始化向量存储对象

    def generate_embeddings(self):
        """生成嵌入向量"""
        documents = self.data_loader.documents
        print(f"正在为{len(documents)}个文档生成嵌入向量...")
        document_embeddings = [self.embeddings.embed(doc.page_content) for doc in documents]
        return document_embeddings

    def create_faiss_vector_store(self):
        """创建FAISS向量存储并保存到本地"""
        # 使用生成的嵌入向量创建FAISS向量存储
        print("开始创建FAISS向量存储...")
        self.vector_store = FAISS.from_documents(self.data_loader.documents, self.embeddings)
        
        # 持久化向量存储到磁盘
        self.vector_store.save_local(self.faiss_index_path)
        print(f"FAISS向量存储已创建并保存至 {self.faiss_index_path}")

    def load_faiss_vector_store(self):
        """加载已保存的FAISS向量存储"""
        if os.path.exists(self.faiss_index_path):
            print(f"加载本地FAISS向量存储：{self.faiss_index_path}")
            self.vector_store = FAISS.load_local(self.faiss_index_path, self.embeddings)
        else:
            print(f"未检测到FAISS向量存储文件：{self.faiss_index_path}，请先创建。")

    def search_similar_documents(self, query, top_k=3):
        """
        根据用户查询执行相似性搜索，并返回最相似的文档内容。
        :param query: 用户查询的文本
        :param top_k: 返回的相似文档数
        :return: 最相似的文档内容列表
        """
        if not self.vector_store:
            print("FAISS向量存储尚未加载，请先创建或加载存储。")
            return []
        
        # 将查询文本嵌入为向量并执行相似性搜索
        query_embedding = self.embeddings.embed(query)
        similar_docs = self.vector_store.similarity_search_by_vector(query_embedding, k=top_k)
        
        # 打印并返回搜索结果
        print(f"\n查询 '{query}' 的相似文档内容：")
        for i, doc in enumerate(similar_docs):
            print(f"文档{i + 1}:\n{doc.page_content[:200]}...\n")  # 显示前200字符
        return [doc.page_content for doc in similar_docs]

# 模块运行测试
if __name__ == "__main__":
    # 初始化数据加载模块
    data_loader = DataLoaderModule(directory_path="./data")
    data_loader.load_all_documents(json_path="./data/sample.json", markdown_path="./data/sample.md")
    
    # 显示数据加载模块的文档概要
    data_loader.display_summary()

    # 初始化嵌入生成与存储模块
    embedding_storage = EmbeddingStorageModule(data_loader)

    # 生成嵌入并创建FAISS向量存储
    embedding_storage.generate_embeddings()
    embedding_storage.create_faiss_vector_store()

    # 加载并执行相似性搜索
    embedding_storage.load_faiss_vector_store()
    query = "文艺复兴的主要贡献"
    embedding_storage.search_similar_documents(query=query, top_k=3)


-------------------------------------------------------------------------------------------------------------


# 导入必要模块
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
import re

# 定义输出解析器的自定义数据结构
class Answer(BaseModel):
    question: str = Field(description="用户提出的问题")
    answer: str = Field(description="对用户问题的简洁回答")
    details: str = Field(description="回答的详细解释")

class PromptEngineeringModule:
    def __init__(self):
        """
        初始化提示词工程模块，包括Prompt模板设计和输出解析器。
        """
        # 初始化OpenAI模型
        self.llm = OpenAI()
        
        # 自定义输出解析器
        self.parser = PydanticOutputParser(pydantic_object=Answer)

    def create_prompt_template(self, query: str) -> str:
        """
        使用Prompt模板生成结构化的模型输入。
        :param query: 用户的查询
        :return: 完整的Prompt文本
        """
        prompt_template = PromptTemplate(
            template=(
                "请回答以下问题，并提供简洁的答案和详细解释：\n"
                "问题：{question}\n\n"
                "格式如下：\n"
                "{format_instructions}\n\n"
                "回答："
            ),
            input_variables=["question"],
            partial_variables={"format_instructions": self.parser.get_format_instructions()}
        )
        
        return prompt_template.format(question=query)

    def parse_response(self, response: str) -> Answer:
        """
        使用自定义输出解析器解析模型的生成响应。
        :param response: 模型生成的文本响应
        :return: 解析后的Answer对象
        """
        return self.parser.parse(response)

    def generate_answer(self, query: str) -> Answer:
        """
        生成企业问答的回答，包含简洁答案和详细解释。
        :param query: 用户的企业问答内容
        :return: Answer对象，包含答案和详细信息
        """
        # 创建Prompt模板
        prompt = self.create_prompt_template(query)
        
        # 模型生成响应
        response = self.llm(prompt)
        print("\n模型原始输出:\n", response)
        
        # 解析响应为结构化数据
        parsed_answer = self.parse_response(response)
        
        # 输出解析结果
        print("\n解析后的结果:")
        print(f"问题: {parsed_answer.question}")
        print(f"简洁回答: {parsed_answer.answer}")
        print(f"详细解释: {parsed_answer.details}")
        
        return parsed_answer

# 模块运行测试
if __name__ == "__main__":
    # 初始化提示词工程模块
    prompt_engineering = PromptEngineeringModule()
    
    # 测试企业问答用例
    query = "请简要说明公司年终绩效考核流程"
    parsed_answer = prompt_engineering.generate_answer(query)


-------------------------------------------------------------------------------------------------------------


# 导入必要模块
from langchain.chains import SequentialChain, LLMChain
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.callbacks import FileCallbackHandler
from typing import Any, Dict

# 任务链调试与监控的回调函数
class DebugCallbackHandler(FileCallbackHandler):
    def __init__(self, log_file="chain_log.txt"):
        """
        初始化调试回调处理程序。
        :param log_file: 日志文件的路径
        """
        self.log_file = log_file
        super().__init__(log_file=self.log_file)

    def on_chain_start(self, chain: Any, inputs: Dict[str, Any]):
        """记录任务链开始的事件"""
        with open(self.log_file, "a") as f:
            f.write(f"\n启动任务链: {chain.__class__.__name__}\n输入: {inputs}\n")

    def on_chain_end(self, outputs: Dict[str, Any]):
        """记录任务链结束的事件"""
        with open(self.log_file, "a") as f:
            f.write(f"输出: {outputs}\n结束任务链。\n")

    def on_chain_error(self, error: Exception):
        """记录任务链中的错误"""
        with open(self.log_file, "a") as f:
            f.write(f"任务链发生错误: {str(error)}\n")

# 任务链模块
class TaskChainModule:
    def __init__(self):
        """
        初始化任务链模块，包含多层任务链和回调调试工具。
        """
        self.llm = OpenAI()
        self.debug_handler = DebugCallbackHandler()

    def create_task_chain(self):
        """构建企业级智能问答的多层任务链"""
        
        # 子任务1：解析用户问题并确定任务类型
        question_template = PromptTemplate(
            template="请分析以下问题，并指出问题的主题和需要的详细信息。\n问题: {query}",
            input_variables=["query"]
        )
        question_chain = LLMChain(llm=self.llm, prompt=question_template, callback_handler=self.debug_handler)

        # 子任务2：根据主题搜索相似内容
        search_template = PromptTemplate(
            template="根据主题'{theme}'，从企业知识库中寻找相关信息，并提供简要概述。",
            input_variables=["theme"]
        )
        search_chain = LLMChain(llm=self.llm, prompt=search_template, callback_handler=self.debug_handler)

        # 子任务3：生成最终的回答格式
        answer_template = PromptTemplate(
            template="将以下信息格式化为企业问答回答:\n问题: {query}\n主题: {theme}\n内容概述: {summary}",
            input_variables=["query", "theme", "summary"]
        )
        answer_chain = LLMChain(llm=self.llm, prompt=answer_template, callback_handler=self.debug_handler)

        # 创建多层任务链
        task_chain = SequentialChain(
            chains=[question_chain, search_chain, answer_chain],
            input_variables=["query"],
            output_variables=["final_answer"],
            callback_handler=self.debug_handler
        )
        return task_chain

    def execute_task_chain(self, query: str):
        """
        执行多层任务链，返回处理后的结果。
        :param query: 用户提出的问题
        :return: 最终回答
        """
        task_chain = self.create_task_chain()
        try:
            result = task_chain({"query": query})
            print("\n任务链执行结果:", result["final_answer"])
            return result["final_answer"]
        except Exception as e:
            print("任务链执行中遇到错误:", str(e))
            return None

# 模块运行测试
if __name__ == "__main__":
    # 初始化任务链模块
    task_module = TaskChainModule()
    
    # 测试企业问答用例
    query = "请解释公司的年度预算规划流程"
    final_answer = task_module.execute_task_chain(query)


-------------------------------------------------------------------------------------------------------------


# 导入必要模块
from langchain.agents import ReActAgent, ZeroShotAgent, AgentExecutor
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate
from langchain.document_loaders import TextLoader
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings

# 定义文档存储模块
class DocumentStorage:
    def __init__(self, directory_path, index_path="faiss_index"):
        """
        初始化文档存储模块，加载文档并创建FAISS向量存储。
        :param directory_path: 企业文档的存储路径
        :param index_path: FAISS向量存储路径
        """
        self.directory_path = directory_path
        self.index_path = index_path
        self.embeddings = OpenAIEmbeddings()
        self.vector_store = self.create_vector_store()
    
    def create_vector_store(self):
        """加载文档并创建FAISS向量存储"""
        loader = TextLoader(self.directory_path)
        documents = loader.load()
        vector_store = FAISS.from_documents(documents, self.embeddings)
        vector_store.save_local(self.index_path)
        return vector_store

    def search_documents(self, query, top_k=3):
        """
        在向量存储中搜索与查询最相似的文档。
        :param query: 查询内容
        :param top_k: 返回的相似文档数量
        :return: 相似文档内容列表
        """
        query_embedding = self.embeddings.embed(query)
        similar_docs = self.vector_store.similarity_search_by_vector(query_embedding, k=top_k)
        
        # 输出文档内容
        print("\n相似文档内容:")
        for i, doc in enumerate(similar_docs):
            print(f"文档{i + 1}:\n{doc.page_content[:200]}...\n")
        
        return [doc.page_content for doc in similar_docs]

# ReAct Agent定义
class ReActAgentSystem:
    def __init__(self, document_storage):
        """
        初始化ReAct Agent，基于企业文档库执行问题解答。
        :param document_storage: 文档存储模块
        """
        self.document_storage = document_storage
        self.llm = OpenAI()
    
    def create_react_prompt(self, query):
        """
        根据查询生成ReAct Prompt。
        :param query: 用户的查询内容
        :return: 生成的Prompt字符串
        """
        react_prompt = PromptTemplate(
            template="请回答以下问题并给出基于企业文档的详细信息:\n问题: {query}\n",
            input_variables=["query"]
        )
        return react_prompt.format(query=query)
    
    def generate_response(self, query):
        """
        使用ReAct Agent生成回答。
        :param query: 用户的查询内容
        :return: ReAct Agent生成的回答
        """
        prompt = self.create_react_prompt(query)
        response = self.llm(prompt)
        print("\nReAct Agent回答:\n", response)
        return response

# Zero-shot Agent定义
class ZeroShotAgentSystem:
    def __init__(self):
        """初始化Zero-shot Agent，适用于不带上下文的直接问答。"""
        self.llm = OpenAI()
    
    def create_zero_shot_prompt(self, query):
        """
        为Zero-shot生成Prompt。
        :param query: 用户的查询内容
        :return: 生成的Prompt字符串
        """
        prompt = PromptTemplate(
            template="请基于常识回答以下问题:\n问题: {query}\n",
            input_variables=["query"]
        )
        return prompt.format(query=query)
    
    def generate_response(self, query):
        """
        使用Zero-shot Agent生成回答。
        :param query: 用户的查询内容
        :return: Zero-shot Agent生成的回答
        """
        prompt = self.create_zero_shot_prompt(query)
        response = self.llm(prompt)
        print("\nZero-shot Agent回答:\n", response)
        return response

# 集成Agent执行系统
class EnterpriseAgentSystem:
    def __init__(self, document_directory):
        """
        初始化企业级Agent系统，结合ReAct Agent和Zero-shot Agent。
        :param document_directory: 企业文档存储路径
        """
        self.document_storage = DocumentStorage(directory_path=document_directory)
        self.react_agent = ReActAgentSystem(document_storage=self.document_storage)
        self.zero_shot_agent = ZeroShotAgentSystem()

    def execute_query(self, query, use_react=True):
        """
        执行企业问答查询，根据查询内容选择适当的Agent。
        :param query: 用户的查询内容
        :param use_react: 是否使用ReAct Agent，默认为True
        :return: 最终回答
        """
        # 使用文档存储库查找相关文档
        related_docs = self.document_storage.search_documents(query=query)
        
        # 根据是否使用ReAct选择Agent
        if use_react:
            print("\n使用ReAct Agent进行回答：")
            response = self.react_agent.generate_response(query)
        else:
            print("\n使用Zero-shot Agent进行回答：")
            response = self.zero_shot_agent.generate_response(query)
        
        print("\n最终回答:", response)
        return response

# 模块运行测试
if __name__ == "__main__":
    # 初始化企业级Agent系统
    document_directory = "./data/enterprise_docs"
    enterprise_agent = EnterpriseAgentSystem(document_directory=document_directory)

    # 测试企业问答用例1：使用ReAct Agent
    query1 = "请说明公司的年度战略计划"
    print("\n查询1结果：")
    enterprise_agent.execute_query(query=query1, use_react=True)
    
    # 测试企业问答用例2：使用Zero-shot Agent
    query2 = "公司的远程办公政策是什么？"
    print("\n查询2结果：")
    enterprise_agent.execute_query(query=query2, use_react=False)


-------------------------------------------------------------------------------------------------------------


# 导入必要模块
from langchain.chains import SequentialChain, LLMChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from typing import Any, Dict
import datetime
import os

# 自定义回调处理程序
class CustomCallbackHandler:
    def __init__(self, log_dir="logs"):
        """
        初始化自定义回调处理程序。
        :param log_dir: 日志文件存储目录
        """
        self.log_dir = log_dir
        os.makedirs(self.log_dir, exist_ok=True)
        self.log_file = os.path.join(self.log_dir, "chain_execution_log.txt")

    def _log_event(self, event: str):
        """将事件记录到日志文件中"""
        with open(self.log_file, "a") as f:
            f.write(f"{datetime.datetime.now()} - {event}\n")

    def on_chain_start(self, chain_name: str, inputs: Dict[str, Any]):
        """记录任务链启动事件"""
        self._log_event(f"启动任务链: {chain_name}\n输入: {inputs}")

    def on_chain_end(self, outputs: Dict[str, Any]):
        """记录任务链结束事件"""
        self._log_event(f"任务链结束\n输出: {outputs}\n")

    def on_chain_error(self, error: Exception):
        """记录任务链错误事件"""
        self._log_event(f"任务链发生错误: {str(error)}\n")

# 任务链设计与调试
class DebuggableTaskChain:
    def __init__(self, callback_handler: CustomCallbackHandler):
        """
        初始化可调试的任务链，集成自定义回调处理程序。
        :param callback_handler: 自定义回调处理程序实例
        """
        self.llm = OpenAI()
        self.callback_handler = callback_handler

    def create_chain(self):
        """构建任务链，用于企业问答系统的分步执行"""

        # 子任务1：问题解析
        question_prompt = PromptTemplate(
            template="请分析以下问题并确定其关键主题。\n问题: {query}",
            input_variables=["query"]
        )
        question_chain = LLMChain(
            llm=self.llm, 
            prompt=question_prompt,
            callback_handler=self.callback_handler
        )

        # 子任务2：文档搜索
        search_prompt = PromptTemplate(
            template="根据主题'{theme}'在企业知识库中搜索信息并总结关键点。",
            input_variables=["theme"]
        )
        search_chain = LLMChain(
            llm=self.llm, 
            prompt=search_prompt,
            callback_handler=self.callback_handler
        )

        # 子任务3：格式化回答
        format_prompt = PromptTemplate(
            template="请根据问题 '{query}' 和找到的内容进行回答，并提供详细说明。\n内容: {summary}",
            input_variables=["query", "summary"]
        )
        format_chain = LLMChain(
            llm=self.llm, 
            prompt=format_prompt,
            callback_handler=self.callback_handler
        )

        # 创建任务链
        task_chain = SequentialChain(
            chains=[question_chain, search_chain, format_chain],
            input_variables=["query"],
            output_variables=["final_answer"],
            callback_handler=self.callback_handler
        )
        
        return task_chain

    def execute_chain(self, query: str):
        """
        执行任务链并记录执行过程。
        :param query: 用户的查询内容
        :return: 最终答案
        """
        task_chain = self.create_chain()
        self.callback_handler.on_chain_start("企业问答任务链", {"query": query})
        
        try:
            result = task_chain({"query": query})
            self.callback_handler.on_chain_end(result)
            print("\n任务链执行结果:", result["final_answer"])
            return result["final_answer"]
        except Exception as e:
            self.callback_handler.on_chain_error(e)
            print("任务链执行中遇到错误:", str(e))
            return None

# 模块运行测试
if __name__ == "__main__":
    # 初始化自定义回调处理程序
    callback_handler = CustomCallbackHandler()

    # 初始化任务链模块
    debuggable_task_chain = DebuggableTaskChain(callback_handler)

    # 测试企业问答用例
    query = "公司的年度增长策略是什么？"
    final_answer = debuggable_task_chain.execute_chain(query=query)


-------------------------------------------------------------------------------------------------------------


# 单元测试1：数据加载模块测试
from unittest import TestCase

class TestDataLoaderModule(TestCase):
    def setUp(self):
        # 初始化数据加载模块
        self.data_loader = DataLoaderModule(directory_path="./test_data")
        # 加载和预处理文档
        self.data_loader.load_all_documents(json_path="./test_data/sample.json", markdown_path="./test_data/sample.md")

    def test_document_count(self):
        """测试数据加载后文档数量"""
        self.assertTrue(len(self.data_loader.documents) > 0, "数据加载失败，文档数应大于0")
        print(f"文档总数: {len(self.data_loader.documents)}")

    def test_document_cleaning(self):
        """测试文档清洗"""
        initial_count = len(self.data_loader.documents)
        self.data_loader.validate_and_clean_data()
        cleaned_count = len(self.data_loader.documents)
        self.assertTrue(cleaned_count <= initial_count, "清洗后文档数量应不大于初始数量")
        print(f"清洗后文档数: {cleaned_count}")



# 单元测试2：嵌入生成与存储模块测试
from unittest import TestCase

class TestEmbeddingStorageModule(TestCase):
    def setUp(self):
        # 初始化嵌入生成与存储模块
        self.data_loader = DataLoaderModule(directory_path="./test_data")
        self.data_loader.load_all_documents(json_path="./test_data/sample.json", markdown_path="./test_data/sample.md")
        self.embedding_storage = EmbeddingStorageModule(data_loader=self.data_loader)

    def test_embedding_generation(self):
        """测试嵌入生成"""
        embeddings = self.embedding_storage.generate_embeddings()
        self.assertEqual(len(embeddings), len(self.data_loader.documents), "生成嵌入数量应等于文档数量")
        print(f"生成的嵌入向量数量: {len(embeddings)}")

    def test_faiss_storage_creation(self):
        """测试FAISS向量存储创建"""
        self.embedding_storage.create_faiss_vector_store()
        self.assertTrue(os.path.exists(self.embedding_storage.faiss_index_path), "FAISS索引文件未创建")
        print("FAISS索引文件已创建")



# 单元测试3：提示词工程模块测试
from unittest import TestCase

class TestPromptEngineeringModule(TestCase):
    def setUp(self):
        # 初始化提示词工程模块
        self.prompt_engineering = PromptEngineeringModule()

    def test_prompt_template_creation(self):
        """测试Prompt模板创建"""
        query = "公司财务审核流程是什么？"
        prompt = self.prompt_engineering.create_prompt_template(query)
        self.assertIn("公司财务审核流程是什么", prompt, "Prompt模板生成错误")
        print("Prompt模板生成成功:", prompt)

    def test_response_parsing(self):
        """测试响应解析"""
        mock_response = '{"question": "公司财务审核流程是什么？", "answer": "财务审核流程包括预算编制、审批和执行。", "details": "详细的审核流程涉及财务部门与项目团队的多轮审批。"}'
        parsed_answer = self.prompt_engineering.parse_response(mock_response)
        self.assertEqual(parsed_answer.question, "公司财务审核流程是什么？", "问题解析错误")
        self.assertEqual(parsed_answer.answer, "财务审核流程包括预算编制、审批和执行。", "简洁回答解析错误")
        print("响应解析成功:", parsed_answer)



# 单元测试4：任务链设计模块测试
from unittest import TestCase

class TestDebuggableTaskChain(TestCase):
    def setUp(self):
        # 初始化任务链模块和回调处理程序
        self.callback_handler = CustomCallbackHandler()
        self.task_chain = DebuggableTaskChain(callback_handler=self.callback_handler)

    def test_chain_execution(self):
        """测试任务链执行"""
        query = "解释年度预算编制流程"
        result = self.task_chain.execute_chain(query)
        self.assertIsNotNone(result, "任务链执行失败")
        print("任务链执行成功，结果:", result)



# 单元测试5：Agent系统模块测试
from unittest import TestCase

class TestEnterpriseAgentSystem(TestCase):
    def setUp(self):
        # 初始化Agent系统
        self.document_directory = "./test_data/enterprise_docs"
        self.enterprise_agent = EnterpriseAgentSystem(document_directory=self.document_directory)

    def test_react_agent_response(self):
        """测试ReAct Agent响应"""
        query = "年度增长策略有哪些内容？"
        response = self.enterprise_agent.execute_query(query=query, use_react=True)
        self.assertIsNotNone(response, "ReAct Agent执行失败")
        print("ReAct Agent响应成功，结果:", response)

    def test_zero_shot_agent_response(self):
        """测试Zero-shot Agent响应"""
        query = "公司的远程办公政策是什么？"
        response = self.enterprise_agent.execute_query(query=query, use_react=False)
        self.assertIsNotNone(response, "Zero-shot Agent执行失败")
        print("Zero-shot Agent响应成功，结果:", response)



# 单元测试6：回调机制与监控模块测试
from unittest import TestCase

class TestCustomCallbackHandler(TestCase):
    def setUp(self):
        # 初始化回调处理程序和任务链
        self.callback_handler = CustomCallbackHandler()
        self.debuggable_task_chain = DebuggableTaskChain(callback_handler=self.callback_handler)

    def test_callback_logging(self):
        """测试回调处理程序的日志记录"""
        query = "公司的年度增长计划是什么？"
        self.debuggable_task_chain.execute_chain(query)
        
        # 验证日志文件是否存在且内容是否正确
        log_file = self.callback_handler.log_file
        self.assertTrue(os.path.exists(log_file), "回调日志文件未创建")
        with open(log_file, "r") as f:
            log_content = f.read()
            self.assertIn("启动任务链", log_content, "日志记录内容错误")
        print("回调日志记录测试成功，日志内容:\n", log_content)



# 导入必要模块和类
import os
from unittest import TestCase

# 集成测试类
class TestEnterpriseQuestionAnsweringSystem(TestCase):
    def setUp(self):
        # 设置数据加载和向量存储路径
        self.document_directory = "./test_data/enterprise_docs"
        
        # 初始化各个模块
        print("初始化数据加载模块...")
        self.data_loader = DataLoaderModule(directory_path=self.document_directory)
        self.data_loader.load_all_documents(json_path="./test_data/sample.json", markdown_path="./test_data/sample.md")

        print("初始化嵌入生成与存储模块...")
        self.embedding_storage = EmbeddingStorageModule(data_loader=self.data_loader)
        self.embedding_storage.create_faiss_vector_store()

        print("初始化提示词工程模块...")
        self.prompt_engineering = PromptEngineeringModule()

        print("初始化任务链模块和回调处理程序...")
        self.callback_handler = CustomCallbackHandler()
        self.task_chain = DebuggableTaskChain(callback_handler=self.callback_handler)

        print("初始化企业级Agent系统...")
        self.enterprise_agent = EnterpriseAgentSystem(document_directory=self.document_directory)

    def test_complete_qa_flow(self):
        """
        集成测试完整的问答流程，从文档加载、嵌入生成到任务链执行和Agent选择。
        """
        # 模拟用户查询
        query = "公司的年度增长策略是什么？"

        # 执行数据加载模块
        print("执行数据加载模块...")
        initial_doc_count = len(self.data_loader.documents)
        self.assertTrue(initial_doc_count > 0, "数据加载失败，文档数应大于0")
        print(f"数据加载模块成功，文档总数: {initial_doc_count}")

        # 执行嵌入生成与存储模块
        print("执行嵌入生成与存储模块...")
        embeddings = self.embedding_storage.generate_embeddings()
        self.assertEqual(len(embeddings), initial_doc_count, "生成的嵌入数量应与文档数量一致")
        print("嵌入生成与存储模块成功")

        # 执行提示词工程模块
        print("执行提示词工程模块...")
        prompt = self.prompt_engineering.create_prompt_template(query)
        self.assertIn("公司的年度增长策略", prompt, "Prompt模板生成错误")
        print("提示词工程模块成功")

        # 执行任务链模块
        print("执行任务链模块...")
        task_chain_result = self.task_chain.execute_chain(query)
        self.assertIsNotNone(task_chain_result, "任务链执行失败")
        print("任务链模块成功，结果:", task_chain_result)

        # 执行Agent系统模块
        print("执行Agent系统模块...")
        agent_response = self.enterprise_agent.execute_query(query=query, use_react=True)
        self.assertIsNotNone(agent_response, "Agent系统执行失败")
        print("Agent系统模块成功，Agent响应:", agent_response)

        # 验证回调日志记录
        print("验证回调日志记录...")
        log_file = self.callback_handler.log_file
        self.assertTrue(os.path.exists(log_file), "回调日志文件未创建")
        with open(log_file, "r") as f:
            log_content = f.read()
            self.assertIn("启动任务链", log_content, "日志记录内容错误")
        print("回调日志记录测试成功，日志内容:\n", log_content)

# 运行集成测试
if __name__ == "__main__":
    import unittest
    unittest.main()


-------------------------------------------------------------------------------------------------------------


from flask import Flask, jsonify, request
import os

# 初始化 Flask 应用
app = Flask(__name__)

# 模块初始化
print("初始化数据加载模块...")
data_loader = DataLoaderModule(directory_path="./data")
data_loader.load_all_documents(json_path="./data/sample.json", markdown_path="./data/sample.md")

print("初始化嵌入生成与存储模块...")
embedding_storage = EmbeddingStorageModule(data_loader=data_loader)
embedding_storage.create_faiss_vector_store()

print("初始化提示词工程模块...")
prompt_engineering = PromptEngineeringModule()

print("初始化任务链模块和回调处理程序...")
callback_handler = CustomCallbackHandler()
task_chain = DebuggableTaskChain(callback_handler=callback_handler)

print("初始化企业级Agent系统...")
enterprise_agent = EnterpriseAgentSystem(document_directory="./data")

# 定义API接口

@app.route("/load_data", methods=["POST"])
def load_data():
    """API接口: 数据加载"""
    directory = request.json.get("directory", "./data")
    json_path = request.json.get("json_path", "./data/sample.json")
    markdown_path = request.json.get("markdown_path", "./data/sample.md")

    # 加载文档
    data_loader.directory_path = directory
    data_loader.load_all_documents(json_path=json_path, markdown_path=markdown_path)
    return jsonify({"status": "success", "message": "数据加载完成", "document_count": len(data_loader.documents)})

@app.route("/generate_embeddings", methods=["POST"])
def generate_embeddings():
    """API接口: 嵌入生成与存储"""
    embeddings = embedding_storage.generate_embeddings()
    embedding_storage.create_faiss_vector_store()
    return jsonify({"status": "success", "message": "嵌入生成并存储完成", "embedding_count": len(embeddings)})

@app.route("/query_chain", methods=["POST"])
def query_chain():
    """API接口: 任务链执行"""
    query = request.json.get("query", "")
    if not query:
        return jsonify({"status": "error", "message": "查询不能为空"}), 400

    result = task_chain.execute_chain(query)
    if result is None:
        return jsonify({"status": "error", "message": "任务链执行失败"}), 500
    return jsonify({"status": "success", "result": result})

@app.route("/agent_query", methods=["POST"])
def agent_query():
    """API接口: 企业级Agent问答"""
    query = request.json.get("query", "")
    use_react = request.json.get("use_react", True)

    if not query:
        return jsonify({"status": "error", "message": "查询不能为空"}), 400

    response = enterprise_agent.execute_query(query=query, use_react=use_react)
    if response is None:
        return jsonify({"status": "error", "message": "Agent执行失败"}), 500
    return jsonify({"status": "success", "response": response})

@app.route("/generate_prompt", methods=["POST"])
def generate_prompt():
    """API接口: 提示词生成"""
    query = request.json.get("query", "")
    if not query:
        return jsonify({"status": "error", "message": "查询不能为空"}), 400

    prompt = prompt_engineering.create_prompt_template(query)
    return jsonify({"status": "success", "prompt": prompt})

@app.route("/get_logs", methods=["GET"])
def get_logs():
    """API接口: 获取任务链执行日志"""
    log_file = callback_handler.log_file
    if not os.path.exists(log_file):
        return jsonify({"status": "error", "message": "日志文件不存在"}), 404

    with open(log_file, "r") as f:
        logs = f.readlines()
    return jsonify({"status": "success", "logs": logs})

# 启动Flask应用
if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=5000)


-------------------------------------------------------------------------------------------------------------


POST /load_data
{
    "directory": "./data",
    "json_path": "./data/sample.json",
    "markdown_path": "./data/sample.md"
}


POST /generate_embeddings


POST /query_chain
{
    "query": "公司的年度增长策略是什么？"
}


POST /agent_query
{
    "query": "公司的远程办公政策是什么？",
    "use_react": false
}


GET /get_logs


-------------------------------------------------------------------------------------------------------------


>> sudo apt install git -y
>> python3 -m venv env
>> source env/bin/activate
>> git clone <REPO_URL>  # 替换为项目的Git仓库地址
>> cd <PROJECT_DIRECTORY>  # 进入项目目录
>> pip install -r requirements.txt


import os
from dotenv import load_dotenv

load_dotenv()  # 加载.env文件


>> export FLASK_APP=app.py  # 指定主应用文件名
>> flask run --host=0.0.0.0 --port=5000
>> gunicorn -w 4 -b 0.0.0.0:5000 app:app
>> sudo apt install nginx -y
>> sudo nano /etc/nginx/sites-available/qa_system


server {
    listen 80;
    server_name <YOUR_SERVER_IP_OR_DOMAIN>;

    location / {
        proxy_pass http://127.0.0.1:5000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}


>> sudo ln -s /etc/nginx/sites-available/qa_system /etc/nginx/sites-enabled
>> sudo nginx -t  # 检查Nginx配置语法
>> sudo systemctl restart nginx
>> sudo ufw allow 'Nginx Full'
>> sudo ufw enable
>> sudo nano /etc/systemd/system/qa_system.service


[Unit]
Description=Gunicorn instance to serve QA System
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/path/to/your/project  # 替换为项目路径
Environment="PATH=/path/to/your/project/env/bin"  # 替换为虚拟环境路径
ExecStart=/path/to/your/project/env/bin/gunicorn -w 4 -b 0.0.0.0:5000 app:app  # Gunicorn启动命令

[Install]
WantedBy=multi-user.target


>> sudo systemctl start qa_system
>> sudo systemctl enable qa_system


-------------------------------------------------------------------------------------------------------------


from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI
from langchain.document_loaders import TextLoader
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.output_parsers import PydanticOutputParser, OutputFixingParser
from pydantic import BaseModel, Field
from typing import List

# 定义数据结构用于解析输出
class Answer(BaseModel):
    question: str = Field(description="用户提出的问题")
    answer: str = Field(description="对用户问题的简洁回答")
    details: str = Field(description="回答的详细解释")

# 自动修复解析器的定义
output_parser = OutputFixingParser(PydanticOutputParser(pydantic_object=Answer))

class EnterpriseQASystem:
    def __init__(self, document_directory: str, index_path="faiss_index"):
        """
        初始化企业问答系统，包括文档存储、MMR搜索和解析器。
        :param document_directory: 企业文档目录
        :param index_path: FAISS向量存储路径
        """
        # 文档加载和FAISS存储初始化
        print("加载文档并创建向量存储...")
        self.embeddings = OpenAIEmbeddings()
        self.document_loader = TextLoader(document_directory)
        self.vector_store = self.create_vector_store()

        # 初始化模型
        self.llm = OpenAI()
        self.index_path = index_path

    def create_vector_store(self):
        """创建FAISS向量存储，用于MMR搜索"""
        documents = self.document_loader.load()
        vector_store = FAISS.from_documents(documents, self.embeddings)
        vector_store.save_local(self.index_path)
        return vector_store

    def mmr_search(self, query: str, top_k: int = 3) -> List[str]:
        """
        使用MMR搜索进行文档检索，减少重复内容并提升结果的多样性。
        :param query: 用户查询
        :param top_k: 返回的文档数量
        :return: 相似文档内容列表
        """
        query_embedding = self.embeddings.embed(query)
        similar_docs = self.vector_store.maximal_marginal_relevance_search_by_vector(query_embedding, k=top_k)
        
        # 打印搜索结果
        print("\nMMR搜索结果:")
        for i, doc in enumerate(similar_docs):
            print(f"文档{i + 1}:\n{doc.page_content[:200]}...\n")
        
        return [doc.page_content for doc in similar_docs]

    def generate_answer(self, query: str) -> Answer:
        """
        生成企业问答的回答，包含简洁答案和详细解释，使用自动修复解析器进行格式化。
        :param query: 用户的企业问答内容
        :return: Answer对象，包含答案和详细信息
        """
        # 使用MMR搜索找到相关文档
        relevant_docs = self.mmr_search(query)
        combined_docs = "\n".join(relevant_docs)

        # Prompt模板定义
        prompt_template = PromptTemplate(
            template=(
                "请回答以下问题，并提供简洁的答案和详细解释：\n"
                "问题: {question}\n"
                "相关内容:\n{context}\n"
                "格式:\n{format_instructions}\n"
                "回答："
            ),
            input_variables=["question", "context"],
            partial_variables={"format_instructions": output_parser.get_format_instructions()}
        )
        
        prompt = prompt_template.format(question=query, context=combined_docs)
        
        # 生成响应并使用自动修复解析器
        raw_response = self.llm(prompt)
        print("\n模型原始输出:\n", raw_response)
        
        # 使用解析器格式化输出
        structured_answer = output_parser.parse(raw_response)
        
        # 输出最终结果
        print("\n解析后的结构化结果:")
        print(f"问题: {structured_answer.question}")
        print(f"简洁回答: {structured_answer.answer}")
        print(f"详细解释: {structured_answer.details}")
        
        return structured_answer

    def validate_output(self, answer: Answer) -> bool:
        """
        输出校验，确保结构化结果符合企业问答要求。
        :param answer: Answer对象
        :return: 校验结果，符合要求返回True
        """
        is_valid = bool(answer.question and answer.answer and answer.details)
        print("\n输出校验结果:", "通过" if is_valid else "不通过")
        return is_valid

# 测试集成运行
if __name__ == "__main__":
    # 初始化企业问答系统
    document_directory = "./data/enterprise_docs"
    qa_system = EnterpriseQASystem(document_directory=document_directory)

    # 测试企业问答用例
    query = "请简要说明公司的年度增长策略"
    
    # 生成回答
    answer = qa_system.generate_answer(query)
    
    # 校验输出
    if qa_system.validate_output(answer):
        print("最终答案通过校验")
    else:
        print("最终答案未通过校验")


-------------------------------------------------------------------------------------------------------------


import time
from statistics import mean

class EnterpriseQATester:
    def __init__(self, document_directory: str):
        """初始化测试环境，包括基础和优化配置"""
        self.qa_system_basic = EnterpriseQASystem(document_directory=document_directory)
        self.qa_system_optimized = EnterpriseQASystem(document_directory=document_directory)

    def basic_query(self, query: str):
        """
        基础配置：普通向量搜索，无自动修复解析器和输出校验
        """
        # 使用普通向量搜索
        relevant_docs = self.qa_system_basic.vector_store.similarity_search(query, k=3)
        combined_docs = "\n".join(doc.page_content for doc in relevant_docs)
        
        prompt_template = PromptTemplate(
            template=(
                "请回答以下问题，并提供简洁的答案和详细解释：\n"
                "问题: {question}\n"
                "相关内容:\n{context}\n"
            ),
            input_variables=["question", "context"]
        )
        prompt = prompt_template.format(question=query, context=combined_docs)
        
        # 调用LLM生成响应
        raw_response = self.qa_system_basic.llm(prompt)
        return raw_response

    def optimized_query(self, query: str):
        """
        优化配置：MMR搜索 + 自动修复解析器 + 输出校验
        """
        # 使用MMR搜索并生成结构化输出
        structured_answer = self.qa_system_optimized.generate_answer(query)
        if not self.qa_system_optimized.validate_output(structured_answer):
            return None
        return structured_answer

    def perform_test(self, query: str, iterations: int = 10):
        """
        执行测试，计算基础配置和优化配置的平均响应时间和成功解析率。
        :param query: 测试查询内容
        :param iterations: 每种配置的测试次数
        """
        basic_times = []
        optimized_times = []
        optimized_success_count = 0

        # 基础配置测试
        print("\n基础配置测试...")
        for i in range(iterations):
            start_time = time.time()
            response = self.basic_query(query)
            end_time = time.time()
            basic_times.append(end_time - start_time)
            print(f"基础配置第{i+1}次查询响应时间: {end_time - start_time:.4f} 秒")

        # 优化配置测试
        print("\n优化配置测试...")
        for i in range(iterations):
            start_time = time.time()
            response = self.optimized_query(query)
            end_time = time.time()
            optimized_times.append(end_time - start_time)
            print(f"优化配置第{i+1}次查询响应时间: {end_time - start_time:.4f} 秒")
            
            # 检查解析器成功输出的结果
            if response:
                optimized_success_count += 1

        # 结果统计
        print("\n测试结果统计:")
        print(f"基础配置平均响应时间: {mean(basic_times):.4f} 秒")
        print(f"优化配置平均响应时间: {mean(optimized_times):.4f} 秒")
        print(f"优化配置成功解析率: {optimized_success_count / iterations * 100:.2f}%")

# 运行测试
if __name__ == "__main__":
    document_directory = "./data/enterprise_docs"
    tester = EnterpriseQATester(document_directory=document_directory)
    query = "请简要说明公司的年度增长策略"

    # 执行测试，定量分析优化效果
    tester.perform_test(query=query, iterations=10)

